package io.github.bigbird0101.datatransfer.collector;


import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import integration.utils.trace.LogUtils;
import io.github.bigbird0101.datatransfer.constants.PluginType;
import io.github.bigbird0101.datatransfer.record.Record;
import io.github.bigbird0101.datatransfer.record.impl.DirtyRecord;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;


@Slf4j
public class StdoutPluginCollector extends io.github.bigbird0101.datatransfer.collector.AbstractTaskPluginCollector {


    private static final int DEFAULT_MAX_DIRTYNUM = 128;

    private String taskId;

    private AtomicInteger maxLogNum = new AtomicInteger(-1);

    private AtomicInteger currentLogNum = new AtomicInteger(0);

    public StdoutPluginCollector() {
    }

    public StdoutPluginCollector(String configuration, Communication communication,
                                 PluginType type) {
        super(configuration, communication, type);
        // TODO: 默认128条
        // 最大日志条数为-1时， 都打印
//        maxLogNum = new AtomicInteger(DEFAULT_MAX_DIRTYNUM);
        JSONObject jsonObject = JSONUtil.parseObj(configuration);
        taskId = jsonObject.getStr("taskId");
    }

    private String formatDirty(final Record dirty, final Throwable t,
                               final String msg) {
        Map<String, Object> msgGroup = new HashMap<String, Object>();

        msgGroup.put("type", super.getPluginType().toString());
        if (StringUtils.isNotBlank(msg)) {
            msgGroup.put("message", msg);
        }
        if (null != t && StringUtils.isNotBlank(t.getMessage())) {
            msgGroup.put("exception", t.getMessage());
        }
        if (null != dirty) {
            msgGroup.put("record", DirtyRecord.asDirtyRecord(dirty)
                    .getColumns());
        }

        return JSONUtil.toJsonStr(msgGroup);
    }

    @Override
    public void collectDirtyRecord(Record dirtyRecord, Throwable t,
                                   String errorMessage) {
        int logNum = currentLogNum.getAndIncrement();
        if (logNum == 0 && t != null) {
            LogUtils.error(log, t);
        }
        // TODO: 打印脏数据，未做其它处理
        if (maxLogNum.intValue() < 0 || currentLogNum.intValue() < maxLogNum.intValue()) {
            LogUtils.warn(log, "脏数据-taskId-" + taskId + ": \n"
                    + this.formatDirty(dirtyRecord, t, errorMessage));
        }

        super.collectDirtyRecord(dirtyRecord, t, errorMessage);
    }
}
